package vertxZeebe.service;

import io.vertx.config.ConfigRetriever;
import io.vertx.config.ConfigRetrieverOptions;
import io.vertx.config.ConfigStoreOptions;
import io.vertx.core.*;
import io.vertx.core.json.JsonObject;
import io.vertx.pgclient.PgConnectOptions;
import io.vertx.pgclient.PgPool;
import io.vertx.serviceproxy.ServiceException;
import io.vertx.sqlclient.*;

import java.time.OffsetDateTime;
import java.util.List;
import java.util.stream.Collector;
import java.util.stream.Collectors;

public class PgService {

    SqlClient sqlClient;
    JsonObject sqlMap;

    public PgService(Vertx vertx) {
        PgPool pgPool = PgPool.pool(vertx,
                new PgConnectOptions(new JsonObject()
                        .put("database", "postgres")
                        .put("user", "postgres")
                        .put("password", "postgres")
                        .put("host", "localhost")
                        .put("port", 5432)),
                new PoolOptions(new JsonObject()
                        .put("maxSize", 1))
        );
        //初始化sql语句
        ConfigStoreOptions fileStore = new ConfigStoreOptions()
                .setType("file")
                .setFormat("yaml")
                .setOptional(false)
                .setConfig(new JsonObject()
                        .put("path", "sql/vertxZeebe.yml"));
        ConfigRetrieverOptions options = new ConfigRetrieverOptions()
                .setScanPeriod(0)
                .setIncludeDefaultStores(false)
                .addStore(fileStore);
        ConfigRetriever.create(vertx, options).getConfig()
                .compose(config -> {
                    for (String name : config.fieldNames()) {
                        JsonObject item = config.getJsonObject(name);
                        item.put("sql", item.getString("sql").trim());
                    }
                    sqlMap = config;
                    return Future.succeededFuture(config);
                });

        //pgPool是sqlClient的实现类
        sqlClient = pgPool;
    }

    public static Collector<Row, ?, List<String>> collectString() {
        return Collectors.mapping(row -> row.getString(0), Collectors.toList());
    }

    public void humanTaskComplete(long jobKey, Handler<AsyncResult<Object>> resultHandler) {
        String sql = sqlMap.getJsonObject("humanTaskComplete").getString("sql");
        Future.future(promise ->
                sqlClient.preparedQuery(sql, Tuple.of(jobKey), ar -> {
                    if (ar.succeeded()) {
                        promise.complete();
                    } else {
                        promise.fail(new ServiceException(400, ar.cause().getMessage(),
                                new JsonObject().put("message", ar.cause().getMessage()))
                        );
                    }
                }))
                .setHandler(resultHandler);
    }

    public void humanTaskClaim(long jobKey, long userId, JsonObject groupMap, Handler<AsyncResult<Object>> resultHandler) {
        String sql = sqlMap.getJsonObject("humanTaskClaim").getString("sql");
        Future.future(promise ->
                sqlClient.preparedQuery(sql, Tuple.of(jobKey, userId, groupMap), ar -> {
                    if (ar.succeeded()) {
                        promise.complete();
                    } else {
                        promise.fail(new ServiceException(400, ar.cause().getMessage(),
                                new JsonObject().put("message", ar.cause().getMessage()))
                        );
                    }
                }))
                .setHandler(resultHandler);
    }

    public void setHumanTaskEndTime(long jobKey, long seconds, Handler<AsyncResult<Object>> resultHandler) {
        String sql = sqlMap.getJsonObject("setHumanTaskEndTime").getString("sql");
        Future.future(promise ->
                sqlClient.preparedQuery(sql, Tuple.of(jobKey, seconds), collectString(), ar -> {
                    if (ar.succeeded()) {
                        promise.complete(ar.result().value());
                    } else {
                        promise.fail(new ServiceException(400, ar.cause().getMessage(),
                                new JsonObject().put("message", ar.cause().getMessage()))
                        );
                    }
                }))
                .setHandler(resultHandler);
    }

    public void assignTask(long jobKey, long userId, long assignee, JsonObject userMap, JsonObject groupMap, Handler<AsyncResult<Object>> resultHandler) {
        String sql = sqlMap.getJsonObject("assignTask").getString("sql");
        Future.future(promise ->
                sqlClient.preparedQuery(sql, Tuple.of(jobKey, userId, assignee, userMap, groupMap), collectString(), ar -> {
                    if (ar.succeeded()) {
                        promise.complete(ar.result().value());
                    } else {
                        promise.fail(new ServiceException(400, ar.cause().getMessage(),
                                new JsonObject().put("message", ar.cause().getMessage()))
                        );
                    }
                }))
                .setHandler(resultHandler);
    }

    /**
     * 将人工任务存入task表
     *
     * @param workflowInstanceKey
     * @param jobKey
     * @param name
     * @param description
     * @param variables
     * @param taskForm
     * @param formFields
     * @param assignee
     * @param candidateGroup
     * @param offsetDateTime
     * @return
     */
    public Future<Void> saveTask(Long workflowInstanceKey, Long jobKey, String name, String description, String variables, String taskForm, String formFields, Long assignee, String candidateGroup, OffsetDateTime offsetDateTime) {
        Promise promise = Promise.promise();
        Tuple tuple = Tuple.of(workflowInstanceKey,
                jobKey,
                name,
                description,
                variables,
                taskForm,
                formFields,
                assignee,
                candidateGroup,
                offsetDateTime
        );

        sqlClient.preparedQuery("INSERT INTO task (workflowInstanceKey,jobKey,name,description,variables,taskForm,formFields,assignee,candidateGroup,endTime) VALUES ($1, $2,$3,$4,$5,$6,$7,$8,$9,$10)", tuple, res -> {
            if (res.succeeded()) {
                System.out.println(String.format("%s流程的%s人工任务插入task表成功", workflowInstanceKey, jobKey));
                promise.complete();
            } else {
                System.out.println("Batch insert task failed " + res.cause());
                promise.fail("Batch insert task failed " + res.cause());
            }
        });
        return promise.future();
    }

    /**
     * 将人工任务存入task_claim表
     *
     * @param workflowInstanceKey
     * @param jobKey
     * @param assignee
     * @param offsetDateTime
     * @return
     */
    public Future<Void> saveTaskClaim(Long workflowInstanceKey, Long jobKey, Long assignee, OffsetDateTime offsetDateTime) {
        Promise promise = Promise.promise();
        Tuple tuple = Tuple.of(workflowInstanceKey,
                jobKey,
                assignee,
                offsetDateTime
        );

        sqlClient.preparedQuery("INSERT INTO task_claim (workflowInstanceKey,jobKey,assignee,endTime) VALUES ($1, $2,$3,$4)", tuple, res -> {
            if (res.succeeded()) {
                System.out.println(String.format("%s流程的%s人工任务插入task_claim表成功", workflowInstanceKey, jobKey));
                promise.complete();
            } else {
                System.out.println("Batch insert task_claim failed " + res.cause());
                promise.fail("Batch insert task_claim failed " + res.cause());
            }
        });
        return promise.future();
    }
}
